package com.example.demo.yzy.controller;
import com.alibaba.fastjson.JSON;
import com.example.demo.exception.ExceptionEnum;
import com.example.demo.exception.ResultCode;
import com.example.demo.exception.throwtype.RunException;
import com.example.demo.yzy.vo.DemoVo;
import com.example.demo.yzy.vo.UpdateVo;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.*;
import org.springframework.validation.annotation.Validated;

import java.util.UUID;
import java.util.concurrent.TimeUnit;
/**
* @description: TODO Demo
* @author 杨镇宇
* @date 2022/4/12 11:12
* @version 1.0
*/
@Api(value = "Demo练习", tags = {" Demo练习"})
@RestController
@Validated
@RequestMapping(value="api/demo")
public class DemoController {
    private static Logger log = LoggerFactory.getLogger(DemoController.class);
    @Autowired
    private RedissonClient redissonClient;
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    @ApiOperation(value = "validation校验 练习")
    @CrossOrigin(origins = "*")
    @PostMapping("/sendSms")
    public String sendSms(@Validated DemoVo vo) {
        return "OK";
    }

    @ApiOperation(value = "redis防重复提交测试")
    @CrossOrigin(origins = "*")
    @PostMapping("/redis")
    public String redis(@Validated UpdateVo vo) {
        //redis防重复提交 调用SETNX操作获取锁，如果返回true ，代表获取锁成功
        Boolean aBoolean = redisTemplate.opsForValue().setIfAbsent(JSON.toJSONString(vo), null, 1, TimeUnit.SECONDS);
        if (aBoolean != null && !aBoolean) {
            ResultCode r = ExceptionEnum.REPETITION_ERROR;
            throw  new RunException(r);
        }
        return "OK";
    }

    /**
     * Redission 分布式锁设计(可重入锁)
     * */
    @ApiOperation(value = "Redission 分布式锁设计(可重入锁)",notes = "Redission 分布式锁设计(可重入锁)")
    @ResponseBody
    @PostMapping("/trylock")
    public String trylock(String key)  {
        String msgValue = "trylock";
        //分布式锁名称
        final String lockName = "redissonTryLock-"+key;
        RLock lock = redissonClient.getLock(lockName);
        try{
            //尝试加锁， 最多等待100秒，上锁后10秒自动解锁
            Boolean aBoolean = lock.tryLock(100,10,TimeUnit.SECONDS);
            if (aBoolean){
                log.info("获取到可重入分布式锁");
                log.info("执行业务逻辑");
            }
            return "ok";
        }catch (Exception e){
            log.error("错误",e);
            return "No";
        }finally {
            /**
             * 注:访问共享资源结束后要释放锁
             * */
            if (lock!= null){
                log.info("释放分布式锁");
                lock.unlock();
                //注在某些严格的业务场景下，也可以调用强制释放分布式锁的方法
                //lock.forceUnlock();
            }
        }
    }
    @ApiOperation(value = "Kafka 解决重复消费测试")
    @CrossOrigin(origins = "*")
    @PostMapping("/kafka")
    public String kafka(@Validated UpdateVo vo) {
        //防止消息丢失
        redisTemplate.opsForValue().set(String.valueOf(vo.getId()),vo);
        UUID uuid = UUID.randomUUID();
        log.info("向kafka发送key:"+uuid.toString());
        kafkaTemplate.send("kafka_test",uuid.toString(),JSON.toJSONString(vo)).addCallback(success -> {
            // 消息发送到的topic
            String topicMsg = success.getRecordMetadata().topic();
            // 消息发送到的分区
            int partition = success.getRecordMetadata().partition();
            // 消息在分区内的offset
            long offset = success.getRecordMetadata().offset();
            log.info("发送消息成功:" + topicMsg + "-" + partition + "-" + offset);
        }, failure -> {
            log.error("发送消息失败:" + failure.getMessage());
        });

        return "OK";
    }



}
